<?php

declare(strict_types=1);
/**
 * This file is part of Hyperf.
 *
 * @link     https://www.hyperf.io
 * @document https://hyperf.wiki
 * @contact  group@hyperf.io
 * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
 */
namespace Hyperfx\ServiceGovernanceNacos\Listener;

use Hyperf\Utils\Codec\Json;
use Hyperfx\ServiceGovernanceNacos\Contract\PipeMessageInterface;
use Hyperf\Framework\Event\OnPipeMessage;
use Hyperf\LoadBalancer\Node;
use Hyperf\Contract\StdoutLoggerInterface;
use Hyperf\Event\Contract\ListenerInterface;
use Hyperf\Process\Event\PipeMessage as UserProcessPipeMessage;
use Hyperf\ServiceGovernance\DriverManager;
use Hyperfx\ServiceGovernanceNacos\NodeSelector\NodeSelector;
use Psr\Container\ContainerInterface;

class OnPipeMessageListener implements ListenerInterface
{
    /**
     * @var ContainerInterface
     */
    protected $container;

    /**
     * @var StdoutLoggerInterface
     */
    protected $logger;

    /**
     * @var DriverManager
     */
    protected $driverManager;


    public function __construct(DriverManager $manager, ContainerInterface $container)
    {
        $this->logger = $container->get(StdoutLoggerInterface::class);
        $this->nodeSelector = $container->get(NodeSelector::class);
    }

    public function listen(): array
    {
        return [
            OnPipeMessage::class,
            UserProcessPipeMessage::class,
        ];
    }

    public function process(object $event): void
    {
        if ($event instanceof OnPipeMessage || $event instanceof UserProcessPipeMessage) {
            $event->data instanceof PipeMessageInterface && $this->onPipeMessage($event->data);
        }
    }
    public function onPipeMessage($message) {
        $data = $message->getData();
        $service = $data[0];
        $items = $data[1];

        $nodes = [];
        foreach ($items as $item) {
            $nodes[] = new Node($item['host'], $item['port'], $item['weight']);
        }

        if (!$this->nodeSelector->has($service)) {
            $this->nodeSelector->register($service, $nodes);
        } else {
            $this->nodeSelector->refresh($service, $nodes);
        }

        $this->logger->debug(sprintf('Process onPipeMessage service node refresh successful'), [
            'service' => $service,
            'nodes' => Json::encode($data)
        ]);
    }
}
